package demo.kafka.stream;

import demo.kafka.stream.config.StreamsConfigFactory;
import demo.kafka.stream.toplogy.WordCountTopology;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.Topology;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Properties;

public class WordCountStreamApp {

    private static final Logger log = LoggerFactory.getLogger(WordCountStreamApp.class);
    private static final String CONFIG_FILE_PATH = "config/kafkaStream/word_count.properties";

    public static void main(String[] args) {
        // 1. Load configuration
        Properties streamsConfig;
        Properties appProps;
        try {
            appProps = StreamsConfigFactory.createConfigFromFile(CONFIG_FILE_PATH);
            streamsConfig = appProps; // In our factory, we put all config in one Properties object for simplicity
            log.info("Configuration loaded successfully from {}", CONFIG_FILE_PATH);
        } catch (IOException e) {
            log.error("Failed to load configuration from {}", CONFIG_FILE_PATH, e);
            return;
        }

        // 2. Build the processing topology
        final Topology topology = WordCountTopology.build();
        log.info("Topology description:\n{}", topology.describe());

        // 3. Create and start the Kafka Streams instance
        final KafkaStreams streams = new KafkaStreams(topology, streamsConfig);

        // Add shutdown hook to respond to SIGTERM and gracefully close the streams application
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

        // Start the stream
        try {
            streams.start();
            log.info("Kafka Streams Word Count application started successfully.");
            System.err.println("----------启动成功--------");
        } catch (Exception e) {
            log.error("Failed to start Kafka Streams application", e);
            System.err.println("Kafka Streams application 启动失败: " + e.getMessage());
            System.exit(1);
        }
    }
}